package cn.dmp.tools

import cn.dmp.utils.ParseUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, DataFrameWriter, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by lfq on 171106
  */

//0 校验参数个数
//1 接受程序参数
//2 创建SparkContext
//3 读取日志数据
//4 对数据进行ETL
//5 将结果写入磁盘
//6 关闭sc

object Text2Parquet {



  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("Text2Parquet").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
   /* conf.set("spark.io.compression.codec","org.apache.spark.io.SnappyCompressionCodec")*/
    conf.set("spark.sql.parquet.compression.codec", "snappy")
    //conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
    val sc: SparkContext = new SparkContext(conf)
    //包装sparkcontext后增强
    val sqlContext= new SQLContext(sc)

    val lines: RDD[String] = sc.textFile("F:\\mrdata\\dmpLog171105\\input")
    //将数据进行整理
    val rowRDD: RDD[Row] = lines.map(_.split(",",-1)).filter(_.length >=85).map(splits => {
//      val splits: Array[String] = line.split(",")
//      if (splits.length != 85) {
//        null
//      }


      val sessionid: String = splits(0)

      val advertisersid: Int = ParseUtils.parseInt(splits(1))
      val adorderid: Int = ParseUtils.parseInt(splits(2))
      val adcreativeid: Int = ParseUtils.parseInt(splits(3))
      val adplatformproviderid: Int = ParseUtils.parseInt(splits(4))

      val sdkversion: String = splits(5)
      val adplatformkey: String = splits(6)
      val putinmodeltype: Int = ParseUtils.parseInt(splits(7))
      val requestmode: Int = ParseUtils.parseInt(splits(8))
      val adprice: Double = ParseUtils.parseDouble(splits(9))
      val adppprice: Double = ParseUtils.parseDouble(splits(10))
      val requestdate: String = splits(11)
      val ip: String = splits(12)
      val appid: String = splits(13)
      val appname: String = splits(14)
      val uuid: String = splits(15)
      val device: String = splits(16)
      val client: Int = ParseUtils.parseInt(splits(17))
      val osversion: String = splits(18)
      val density: String = splits(19)
      val pw: Int = ParseUtils.parseInt(splits(20))
      val ph: Int = ParseUtils.parseInt(splits(21))
      val long: String = splits(22)
      val lat: String = splits(23)
      val provincename: String = splits(24)
      val cityname: String = splits(25)
      val ispid: Int =ParseUtils.parseInt(splits(26))
      val ispname: String = splits(27)
      val networkmannerid: Int = ParseUtils.parseInt(splits(28))
      val networkmannername: String = splits(29)
      val iseffective: Int = ParseUtils.parseInt(splits(30))
      val isbilling: Int = ParseUtils.parseInt(splits(31))
      val adspacetype: Int = ParseUtils.parseInt(splits(32))
      val adspacetypename: String = splits(33)
      val devicetype: Int = ParseUtils.parseInt(splits(34))
      val processnode: Int = ParseUtils.parseInt(splits(35))
      val apptype: Int = ParseUtils.parseInt(splits(36))
      val district: String = splits(37)
      val paymode: Int = ParseUtils.parseInt(splits(38))
      val isbid: Int = ParseUtils.parseInt(splits(39))
      val bidprice: Double = ParseUtils.parseDouble(splits(40))
      val winprice: Double = ParseUtils.parseDouble(splits(41))
      val iswin: Int = ParseUtils.parseInt(splits(42))
      val cur: String = splits(43)
      val rate: Double = ParseUtils.parseDouble(splits(44))
      val cnywinprice: Double = ParseUtils.parseDouble(splits(45))
      val imei: String = splits(46)
      val mac: String = splits(47)
      val idfa: String = splits(48)
      val openudid: String = splits(49)
      val androidid: String = splits(50)
      val rtbprovince: String = splits(51)
      val rtbcity: String = splits(52)
      val rtbdistrict: String = splits(53)
      val rtbstreet: String = splits(54)
      val storeurl: String = splits(55)
      val realip: String = splits(56)
      val isqualityapp: Int = ParseUtils.parseInt(splits(57))
      val bidfloor: Double = ParseUtils.parseDouble(splits(58))
      val aw: Int = ParseUtils.parseInt(splits(59))
      val ah: Int = ParseUtils.parseInt(splits(60))
      val imeimd5: String = splits(61)
      val macmd5: String = splits(62)
      val idfamd5: String = splits(63)
      val openudidmd5: String = splits(64)
      val androididmd5: String = splits(65)
      val imeisha1: String = splits(66)
      val macsha1: String = splits(67)
      val idfasha1: String = splits(68)
      val openudidsha1: String = splits(69)
      val androididsha1: String = splits(70)
      val uuidunknow: String = splits(71)
      val userid: String = splits(72)
      val iptype: Int = ParseUtils.parseInt(splits(73))
      val initbidprice: Double = ParseUtils.parseDouble(splits(74))
      val adpayment: Double = ParseUtils.parseDouble(splits(75))
      val agentrate: Double =ParseUtils.parseDouble(splits(76))
      val lomarkrate: Double = ParseUtils.parseDouble(splits(77))
      val adxrate: Double = ParseUtils.parseDouble(splits(78))
      val title: String = splits(79)
      val keywords: String = splits(80)
      val tagid: String = splits(81)
      val callbackdate: String = splits(82)
      val channelid: String = splits(83)
      val mediatype: Int = ParseUtils.parseInt(splits(84))


      Row(sessionid, advertisersid, adorderid, adcreativeid, adplatformproviderid, sdkversion, adplatformkey, putinmodeltype, requestmode,
        adprice, adppprice, requestdate, ip, appid, appname, uuid, device, client, osversion, density, pw, ph,
        long, lat, provincename,
        cityname, ispid, ispname, networkmannerid, networkmannername, iseffective, isbilling, adspacetype, adspacetypename, devicetype,
        processnode, apptype, district, paymode, isbid, bidprice, winprice, iswin, cur, rate, cnywinprice, imei, mac, idfa,
        openudid, androidid, rtbprovince, rtbcity, rtbdistrict, rtbstreet, storeurl, realip, isqualityapp, bidfloor, aw,
        ah, imeimd5, macmd5, idfamd5, openudidmd5, androididmd5, imeisha1, macsha1, idfasha1, openudidsha1, androididsha1, uuidunknow, userid,
        iptype,
        initbidprice,
        adpayment,
        agentrate,
        lomarkrate,
        adxrate, title, keywords, tagid, callbackdate, channelid, mediatype)

    })


    val schema: StructType = StructType(List(
      StructField("sessionid", StringType, true),
      StructField("advertisersid", IntegerType, true),
      StructField("adorderid", IntegerType, true),
      StructField("adcreativeid", IntegerType, true),
      StructField("adplatformproviderid", IntegerType, true),
      StructField("sdkversion", StringType, true),
      StructField("adplatformkey", StringType, true),
      StructField("putinmodeltype", IntegerType, true),
      StructField("requestmode", IntegerType, true),
      StructField("adprice", DoubleType, true),
      StructField("adppprice", DoubleType, true),
      StructField("requestdate",StringType, true),
      StructField("ip",StringType, true),
      StructField("appid",StringType, true),
      StructField("appname",StringType, true),
      StructField("uuid",StringType, true),
      StructField("device",StringType, true),
      StructField("client", IntegerType, true),
      StructField("osversion",StringType, true),
      StructField("density",StringType, true),
      StructField("pw", IntegerType, true),
      StructField("ph", IntegerType, true),
      StructField("long",StringType, true),
      StructField("lat",StringType, true),
      StructField("provincename",StringType,true),
      StructField("cityname",StringType,true),
      StructField("ispid",IntegerType,true),
      StructField("ispname",StringType,true),
      StructField("networkmannerid",IntegerType,true),
      StructField("networkmannername",StringType,true),
      StructField("iseffective",IntegerType,true),
      StructField("isbilling",IntegerType,true),
      StructField("adspacetype",IntegerType,true),
      StructField("adspacetypename",StringType,true),
      StructField("devicetype",IntegerType,true),
      StructField("processnode",IntegerType,true),
      StructField("apptype",IntegerType,true),
      StructField("district",StringType,true),
      StructField("paymode",IntegerType,true),
      StructField("isbid",IntegerType,true),
      StructField("bidprice",DoubleType,true),
      StructField("winprice",DoubleType,true),
      StructField("iswin",IntegerType,true),
      StructField("cur",StringType,true),
      StructField("rate",DoubleType,true),
      StructField("cnywinprice",DoubleType,true),
      StructField("imei",StringType,true),
      StructField("mac",StringType,true),
      StructField("idfa",StringType,true),
      StructField("openudid",StringType,true),
      StructField("androidid",StringType,true),
      StructField("rtbprovince",StringType,true),
      StructField("rtbcity",StringType,true),
      StructField("rtbdistrict",StringType,true),
      StructField("rtbstreet",StringType,true),
      StructField("storeurl",StringType,true),
      StructField("realip",StringType,true),
      StructField("isqualityapp",IntegerType,true),
      StructField("bidfloor",DoubleType,true),
      StructField("aw",IntegerType,true),
      StructField("ah",IntegerType,true),
      StructField("imeimd5",StringType,true),
      StructField("macmd5",StringType,true),
      StructField("idfamd5",StringType,true),
      StructField("openudidmd5",StringType,true),
      StructField("androididmd5",StringType,true),
      StructField("imeisha1",StringType,true),
      StructField("macsha1",StringType,true),
      StructField("idfasha1",StringType,true),
      StructField("openudidsha1",StringType,true),
      StructField("androididsha1",StringType,true),
      StructField("uuidunknow",StringType,true),
      StructField("userid",StringType,true),
      StructField("iptype",IntegerType,true),
      StructField("initbidprice",DoubleType,true),
      StructField("adpayment",DoubleType,true),
      StructField("agentrate",DoubleType,true),
      StructField("lomarkrate",DoubleType,true),
      StructField("adxrate",DoubleType,true),
      StructField("title",StringType,true),
      StructField("keywords",StringType,true),
      StructField("tagid",StringType,true),
      StructField("callbackdate",StringType,true),
      StructField("channelid",StringType,true),
      StructField("mediatype",IntegerType,true)
    ))

    //rdd关联scheme
    val bdf: DataFrame = sqlContext.createDataFrame(rowRDD,schema)


   //分区

    //保存为parquet格式

    bdf.write.parquet("F:\\mrdata\\dmpLog171105\\DSP_Log_Parquet")
//    val dfw: DataFrameWriter = new DataFrameWriter(bdf)
//    dfw.format("parquet").save("F:\\mrdata\\dmpLog171105\\DSP_Log_Parquet2")

    sc.stop()
  }
}
